package com.yifeng.repo.flink.data.transport.streaming.connectors.tidb.sink;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Objects;

import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.util.StringUtils;
import com.alibaba.fastjson.JSONObject;
import com.google.common.base.Function;
import com.google.common.collect.Lists;
import com.yifeng.repo.flink.data.transport.config.SinkDbConfig;
import com.yifeng.repo.flink.data.transport.dto.DataRow;
import com.yifeng.repo.flink.data.transport.relational.connection.DruidDataSourceManager;
import com.yifeng.repo.flink.data.transport.relational.connection.DruidJdbcExecutor;

/**
 * ClassName: WriteToTidbSink
 * Description: 写入目标数据库sink
 * @since 2023-06-18
 */
public class WriteToTidbSink extends RichSinkFunction<String> {
	private static final long serialVersionUID = 1L;

	private static final Logger log = LoggerFactory.getLogger(WriteToTidbSink.class);
	
	private SinkDbConfig sinkDbConfig;
	
	private DruidDataSource dataSource;
	
	private DruidJdbcExecutor druidJdbcExecutor;
	
	private List<String> keyword;
	
	private List<String> needDealKeywordTable;
	
	public WriteToTidbSink(SinkDbConfig sinkDbConfig) {
		this.sinkDbConfig = sinkDbConfig;
		if(!StringUtils.isEmpty(sinkDbConfig.getKeyWord())) {
			this.keyword = Arrays.asList(sinkDbConfig.getKeyWord().split(","));
		}
		if(!StringUtils.isEmpty(sinkDbConfig.getNeedDealKeywordTable())) {
			this.needDealKeywordTable = Arrays.asList(sinkDbConfig.getNeedDealKeywordTable().split(","));
		}
	}
	
    @Override
    public void open(Configuration parameters) throws Exception {
    	DruidDataSourceManager.Config oracleConfig = new DruidDataSourceManager.Config(sinkDbConfig.getJdbcUrl(),sinkDbConfig.getUsername(), sinkDbConfig.getPassword(),sinkDbConfig.getPoolSize());
        this.dataSource = DruidDataSourceManager.getDataSource(oracleConfig);
        this.druidJdbcExecutor = new DruidJdbcExecutor();
    }
	
	@Override
	public void invoke(String data, Context context) throws Exception {
		List<DataRow> dataRowList = JSONObject.parseArray(data, DataRow.class);
		if(CollectionUtils.isEmpty(dataRowList)) {
			return;
		}
		DataRow dataRowFirst = dataRowList.get(0);
		if(Objects.isNull(dataRowFirst)) {
			return;
		}
		//ddl这期版本先不处理2023-06-18
        if(DataRow.Type.end.equals(dataRowFirst.getOp()) || DataRow.Type.ddl.equals(dataRowFirst.getOp())) {
        	log.info("操作类型为end或者ddl: {}",dataRowFirst.getOp());
        	return;
		}
        String sourceTableName = transToLowerCase(dataRowList.get(0).getTb());
        String sourceSchemaName = transToLowerCase(dataRowList.get(0).getSchema());
        Map<String,String> tableMappingMap = sinkDbConfig.getTableMappingMap();
        String keyTable = sourceTableName;
        if(!StringUtils.isEmpty(sourceSchemaName)) {
        	keyTable = sourceSchemaName + "." + sourceTableName;
        }
        String tableName = tableMappingMap.get(keyTable);
        if(StringUtils.isEmpty(tableName)) {
        	tableName = sourceTableName;
        }
		//主键信息
		Map<String,String> pkMap = sinkDbConfig.getPkMap();
		//存在多主键
		String pkColumn = pkMap.get(sourceTableName);
		if(StringUtils.isEmpty(pkColumn)) {
			close();
			throw new RuntimeException("表的主键未配置：" + sourceTableName);
		}
		String[] pkColumns = pkColumn.split(",");
		List<String> pkColumnList = Arrays.asList(pkColumns);
        //判断操作类型
        if(DataRow.Type.d.equals(dataRowFirst.getOp())) {
        	//使用before的数据
        	List<Map<String,Object>> deleteDataList = new ArrayList<Map<String,Object>>();
    	    for(DataRow dataRow : dataRowList) {
    	    	if(Objects.nonNull(dataRow.getBefore())) {
    	    		deleteDataList.add(dataRow.getBefore());
    	    	}
    	    }
    	    if(CollectionUtils.isEmpty(deleteDataList)) {
    	    	return;
    	    }
        	List<List<Object>> deleteList = assembleValueListForDelete(deleteDataList,pkColumnList);
        	druidJdbcExecutor.updateList(dataSource, assembleDeleteSql(tableName,transferKeywordGetNew(pkColumnList,sourceTableName)), deleteList);
        	return;
        }
        //将source批量插入或者更新
        List<String> fieldKeyList = new ArrayList<String>();
        for (String key: dataRowFirst.getAfter().keySet()){
        	fieldKeyList.add(key.toLowerCase());
        }
        //转换关键字
        fieldKeyList = transferKeywordGetNew(fieldKeyList,sourceTableName);
        List<Map<String,Object>> dataList = new ArrayList<Map<String,Object>>();
	    for(DataRow dataRow : dataRowList) {
	       dataList.add(dataRow.getAfter());
	    }
        if(DataRow.Type.r.equals(dataRowFirst.getOp())) {
        	druidJdbcExecutor.insertList(dataSource, tableName,fieldKeyList, assembleValueList(dataList));
        }
        if(DataRow.Type.c.equals(dataRowFirst.getOp())) {
        	druidJdbcExecutor.insertOrUpdateByPk(dataSource, tableName,fieldKeyList, assembleValueList(dataList),transferKeywordGetNew(pkColumnList,sourceTableName).toArray(new String[0]));
        }
        if(DataRow.Type.u.equals(dataRowFirst.getOp())) {
        	druidJdbcExecutor.updateList(dataSource, assembleUpdateSql(tableName,fieldKeyList,transferKeywordGetNew(pkColumnList,sourceTableName)), assembleValueListForUpdate(dataList,pkColumnList));
        }
	}
	
	@Override
    public void close() throws Exception {
		log.info("结束写入");
		dataSource.close();
    }
	
	private String transToLowerCase(String value) {
		if(StringUtils.isEmpty(value)) {
			return value;
		}
		return value.toLowerCase();
	}
	
	/**
	 * 组装更新的sql语句
	 * @param tableName 表名
	 * @param fieldKeyList 字段名列表
	 * @param pkColumnList 主键名列表
	 * @return 更新语句
	 */
	private String assembleUpdateSql(String tableName,List<String> fieldKeyList,List<String> pkColumnList) {
		String updateSql = "update " + tableName + " set ";
		for(String fieldKey : fieldKeyList) {
			if(isExistData(pkColumnList,fieldKey)) {
				continue;
			}
			updateSql = updateSql + fieldKey + " =?,";
		}
		updateSql = updateSql.substring(0,updateSql.length() -1);
		updateSql = updateSql + " where ";
		for(String pkColumn : pkColumnList) {
			updateSql = updateSql + pkColumn + " =? and ";
		}
		updateSql = updateSql.substring(0,updateSql.length() -4);
		return updateSql;
	}
	
	/**
	 * 组装更新的sql语句
	 * @param tableName 表名
	 * @param fieldKeyList 字段名列表
	 * @param pkColumnList 主键名列表
	 * @return 更新语句
	 */
	private String assembleDeleteSql(String tableName,List<String> pkColumnList) {
		String deleteSql = "delete from " + tableName;
		deleteSql = deleteSql + " where ";
		for(String pkColumn : pkColumnList) {
			deleteSql = deleteSql + pkColumn + " =? and ";
		}
		deleteSql = deleteSql.substring(0,deleteSql.length() -4);
		return deleteSql;
	}
	
	/**
	 * 组装数据-针对新增类型
	 * @param sourceList 源数据
	 * @return 去除value值
	 */
	private List<List<Object>> assembleValueList(List<Map<String,Object>> sourceList) {
    	if(CollectionUtils.isEmpty(sourceList)) {
    		return new ArrayList<List<Object>>();
    	}
    	List<List<Object>> dataList = Lists.transform(sourceList, new Function<Map<String,Object>,List<Object>>(){
			@Override
			public List<Object> apply(Map<String,Object> map) {
				List<Object> row = new ArrayList<Object>();
				for (String key: map.keySet()){
					row.add(map.get(key));
		        }
				return row;
			}
    	});
		return dataList;
    }
	
	/**
	 * 组装数据-针对新增类型
	 * @param sourceList 源数据
	 * @return 去除value值
	 */
	private List<List<Object>> assembleValueListForDelete(List<Map<String,Object>> sourceList,List<String> pkColumnList) {
    	if(CollectionUtils.isEmpty(sourceList)) {
    		return new ArrayList<List<Object>>();
    	}
    	List<List<Object>> dataList = Lists.transform(sourceList, new Function<Map<String,Object>,List<Object>>(){
			@Override
			public List<Object> apply(Map<String,Object> map) {
				List<Object> row = new ArrayList<Object>();
				for (String key: map.keySet()){
					if(isExistData(pkColumnList,key)) {
						row.add(map.get(key));
					}
		        }
				return row;
			}
    	});
		return dataList;
    }
	
	/**
	 * 组装数据-针对更新类型
	 * @param sourceList 源数据
	 * @return 去除value值
	 */
	private List<List<Object>> assembleValueListForUpdate(List<Map<String,Object>> sourceList,List<String> pkColumnList) {
    	if(CollectionUtils.isEmpty(sourceList)) {
    		return new ArrayList<List<Object>>();
    	}
    	List<List<Object>> dataList = Lists.transform(sourceList, new Function<Map<String,Object>,List<Object>>(){
			@Override
			public List<Object> apply(Map<String,Object> map) {
				List<Object> row = new ArrayList<Object>();
				//最后排主键
				for (String key: map.keySet()){
					if(isExistData(pkColumnList,key)) {
						continue;
					}
					row.add(map.get(key));
		        }
				pkColumnList.forEach(pkColumn -> {
					row.add(map.get(pkColumn.toUpperCase()));
				});
				return row;
			}
    	});
		return dataList;
    }
	
	/**
	 * 判断数据是否存在
	 * @param dataList 数据列表
	 * @param valideData 需要校验的数据
	 * @return true : 数据存在， false ；数据不存在
	 */
	private boolean isExistData(List<String> dataList, String valideData) {
		if(CollectionUtils.isEmpty(dataList)) {
			return false;
		}
		for(String data : dataList) {
			if(valideData.equalsIgnoreCase(data)) {
				return true;
			}
		}
		return false;
	}
	
	private List<String> transferKeywordGetNew(List<String> fieldList,String tableName) {
		if(!sinkDbConfig.isNeedTransferKeyWord() || CollectionUtils.isEmpty(needDealKeywordTable) || !needDealKeywordTable.contains(tableName)) {
			return fieldList;
		}
		List<String> newFieldList = new ArrayList<String>();
		//转换关键字，在后面增加一个特定字符
		fieldList.forEach(field -> {
			if(keyword.contains(field)){
				newFieldList.add(field + sinkDbConfig.getKeyWordSupplement());
			} else {
				newFieldList.add(field);
			}
		});
		return newFieldList;
	}
}
